package com.stormragetech.sylwanas.process;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.GetResponse;
import com.stormragetech.sylwanas.common.MessageWrapper;
import com.stormragetech.sylwanas.common.QueueInfo;
import com.stormragetech.sylwanas.common.ReturnInfo;
import com.stormragetech.sylwanas.constant.Constants;
import com.stormragetech.sylwanas.converter.FastjsonMessageConverter;
import com.stormragetech.sylwanas.converter.ProtobufMessageConverter;
import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.amqp.support.converter.MessageConverter;

/**
 * description: //TODO
 *
 * @author: Xue Bin
 * @version: 1.0.0-SNAPSHOT
 */
@Log4j2
public class MessageConsumerBuilder implements MessageConsumer {

    private QueueInfo info;
    private ConnectionFactory factory;
    private RetryCache cache;
    private int maxCount;
    private MessageHandler handler;
    private final MessagePropertiesConverter mpConverter = new DefaultMessagePropertiesConverter();

    MessageConsumerBuilder(QueueInfo info, ConnectionFactory factory, RetryCache cache, int maxCount, boolean spring) {
        this.info = info;
        this.factory = factory;
        this.cache = cache;
        this.maxCount = maxCount;
        try {
            if (spring) {
                Object obj = cache.getContext().getBean(Class.forName(info.getHandler()));
                if (obj instanceof MessageHandler) {
                    handler = (MessageHandler) obj;
                } else {
                    log.error("业务处理类加载错误");
                }
            } else {
                handler = (MessageHandler) Class.forName(info.getHandler()).newInstance();
            }
        } catch (Exception e) {
            log.error("handler init err: {}", e);
            handler = null;
        }
    }

    @Override
    public ReturnInfo consume() {

        Connection conn = factory.createConnection();
        Channel channel = conn.createChannel(false);

        try {
            GetResponse response = channel.basicGet(info.getQueue(), false);
            while (null == response) {
                response = channel.basicGet(info.getQueue(), false);
                Thread.sleep(Constants.ONE_SECOND);
            }

            Message message = new Message(response.getBody(),
                    mpConverter.toMessageProperties(response.getProps(), response.getEnvelope(), "UTF-8"));
            String id = message.getMessageProperties().getMessageId();
            boolean reQueue = true;
            if (null != cache.getCMap().get(id) && cache.getCMap().get(id).getCount().intValue() > maxCount) {
                reQueue = false;
            }

            @SuppressWarnings("unchecked")
            ReturnInfo returnInfo = execute(message, handler);
            if (Constants.SUCC.equals(returnInfo.getCode())) {
                channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
                System.out.println();
                System.out.println("---------- ---------- ---------- ---------- ---------- ----------");
                log.debug("[ SUCCESS ] - [ COMSUMER ] - [ MESSAGE - ID ] - [ {} ]", id);
                System.out.println("---------- ---------- ---------- ---------- ---------- ----------");
                System.out.println();
            } else {
                System.out.println();
                System.out.println("---------- ---------- ---------- ---------- ---------- ----------");
                log.debug("[ FAILURE ] - [ COMSUMER ] - [ MESSAGE - ID ] - [ {} ] - [ RETURN INFO ] - [ {} ]", id, returnInfo.getMsg());
                System.out.println("---------- ---------- ---------- ---------- ---------- ----------");
                System.out.println();
                channel.basicNack(response.getEnvelope().getDeliveryTag(), false, reQueue);
                if (!reQueue) {
                    // TODO: save err message to file
                    log.error("save err message to file !!!");
                }
            }

            return returnInfo;

        } catch (Exception e) {

            log.error("exception", e);
            return new ReturnInfo(Constants.FAIL, "exception: " + e.getMessage());
        }
    }

    @SuppressWarnings("unchecked")
    private <T> ReturnInfo execute(Message message, MessageHandler<T> handler) {

        // init converter
        //
        try {
            final MessageConverter converter;
            if (MessageProperties.CONTENT_TYPE_JSON.equals(message.getMessageProperties().getContentType())) {
                converter = new FastjsonMessageConverter<T>();
            } else {
                converter = new ProtobufMessageConverter();
            }
            T t = (T) converter.fromMessage(message);
            MessageWrapper wrapper = new MessageWrapper(message.getMessageProperties().getMessageId(), t, this);
            if (null == cache.getCMap().get(wrapper.getId())) {
                cache.add(wrapper, Constants.CONSUME_MSG_WRAPPER);
            }
            return handler.handle(t);
        } catch (Exception e) {
            return new ReturnInfo(Constants.MSG_HANDLE_ERR_C, Constants.MSG_HANDLE_ERR_M + e.getMessage());
        }
    }
}
